package com.yyx.demo.publishSubscribe;

import com.yyx.demo.queue.ZKDistributedQueue;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

public class PublishSubscriber {

    private ZooKeeper zookeeper;
    private List<ACL> acl = ZooDefs.Ids.OPEN_ACL_UNSAFE;
    private static final Logger LOG = LoggerFactory.getLogger(PublishSubscriber.class);
    private SubscribeWatcher subscribeWatcher;
    private String sessionId ;

    public PublishSubscriber(ZooKeeper zooKeeper, List<ACL> acl){
        if (acl != null) {
            this.acl = acl;
        }
        this.zookeeper = zooKeeper;
        this.sessionId="0x" + Long.toHexString(zookeeper.getSessionId());;
        System.out.println(zooKeeper.getSessionId());
        //订阅事件
        subscribeWatcher=new SubscribeWatcher(this.zookeeper,this.acl,sessionId);
    }

    //发布事件
    public boolean PublishEvent(byte[] data,String dir,List<ACL> acl){
        SequenceZnode sequenceZnode = new SequenceZnode(zookeeper,acl);
        try {
            return sequenceZnode.offer(data,dir);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }

    //订阅并消费事件
    public byte[] SubscribeEvent(String dir) throws KeeperException, InterruptedException {
        //判断该dir是否含有
        try {
            Stat stat=zookeeper.exists(dir,null);
            if(stat==null){
                LOG.warn("该dir不存在或者已经被删除");
            }else {
                //订阅事件
                subscribeWatcher.setDir(dir);
                zookeeper.getChildren(dir,subscribeWatcher);
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


        //消费事件
        String sessionName = "/"+this.sessionId;
        String path=sessionName+dir;
        ZKDistributedQueue zkDistributedQueue = new ZKDistributedQueue(zookeeper,path,acl);

        //判断 sessionid是否含有
        Stat stat=zookeeper.exists(sessionName,null);
        if(stat==null){
            zookeeper.create(sessionName, new byte[0], acl, CreateMode.PERSISTENT);
        }
        synchronized (this){
                //查看分布式队列里，有没有值
            try {
                return zkDistributedQueue.takeSync();
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return null;
    }
}
